fix: Filter subscription events by collection before opening txn - #4909
fix: Filter subscription events by collection before opening txn#4909edjroz wants to merge 5 commits into
Conversation
Adds CheckCollectionFilter to *request.Select, resolves the target collection's root CollectionID once at subscribe-time, and rejects events from other collections in the existing docID/cid filter site before opening a transaction. Fixes sourcenetwork#4896
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
📝 WalkthroughWalkthroughAdds collection-level event filtering for subscriptions: Select gains a targetCollectionID field and CheckCollectionFilter method; subscriptionSelector now requires CheckCollectionFilter; a new targetCollectionSetter interface enables stamping at subscribe-time; handleSubscription resolves and stamps the collection; the event loop skips mismatched-collection events. ChangesSubscription collection-level event filtering
Sequence DiagramsequenceDiagram
participant Subscriber
participant handleSubscription
participant Selector as Selector (Select)
participant EventLoop
Subscriber->>handleSubscription: subscribe with Select
handleSubscription->>handleSubscription: resolve collection by name
handleSubscription->>Selector: SetTargetCollectionID(rootCollectionID)
EventLoop->>Selector: CheckCollectionFilter(evt.CollectionID)
Selector-->>EventLoop: true/false (accept/skip event)
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~20 minutes Assessment against linked issues
Possibly related PRs
Suggested labels
Suggested reviewers
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
There was a problem hiding this comment.
🧹 Nitpick comments (1)
internal/db/subscription_collection_filter_test.go (1)
79-98: ⚡ Quick winMake the “no response” assertion deterministic over a time window.
Line 97 uses an instantaneous
defaultcheck, so a slightly delayed wrong-collection response can slip through undetected. Prefer a bounded wait loop that continuously asserts both “no message” and “txn counter unchanged.”Proposed test-shape adjustment
- mid := db.previousTxnID.Load() - time.Sleep(200 * time.Millisecond) - after := db.previousTxnID.Load() - - require.Equal(t, mid, after, - "subscription must not open a transaction for a wrong-collection event; counter advanced from %d to %d", - mid, after) - - // Belt-and-braces: also confirm nothing surfaces on the response channel. - select { - case got, ok := <-subCh: - if !ok { - t.Fatalf("subscription channel closed unexpectedly") - } - t.Fatalf("expected no response for wrong-collection event, got %+v", got) - default: - } + mid := db.previousTxnID.Load() + deadline := time.After(300 * time.Millisecond) + ticker := time.NewTicker(20 * time.Millisecond) + defer ticker.Stop() + + for { + select { + case got, ok := <-subCh: + if !ok { + t.Fatalf("subscription channel closed unexpectedly") + } + t.Fatalf("expected no response for wrong-collection event, got %+v", got) + case <-ticker.C: + current := db.previousTxnID.Load() + require.Equal(t, mid, current, + "subscription must not open a transaction for a wrong-collection event; counter advanced from %d to %d", + mid, current) + case <-deadline: + return + } + }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@internal/db/subscription_collection_filter_test.go` around lines 79 - 98, The instantaneous default select can miss slightly delayed messages; replace it with a bounded wait that repeatedly (or for a fixed duration) asserts both that db.previousTxnID.Load() remains equal to mid and that no message is received on subCh—for example, use a time.After timeout (e.g., 200ms) with a loop/select that checks case <-subCh (fail if received) and case <-time.After(shortInterval) to re-check the txn counter until the overall timeout expires, then succeed if no messages arrived and the counter stayed unchanged; reference the variables mid, after (or directly db.previousTxnID.Load()) and the channel subCh in the updated logic.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@internal/db/subscription_collection_filter_test.go`:
- Around line 79-98: The instantaneous default select can miss slightly delayed
messages; replace it with a bounded wait that repeatedly (or for a fixed
duration) asserts both that db.previousTxnID.Load() remains equal to mid and
that no message is received on subCh—for example, use a time.After timeout
(e.g., 200ms) with a loop/select that checks case <-subCh (fail if received) and
case <-time.After(shortInterval) to re-check the txn counter until the overall
timeout expires, then succeed if no messages arrived and the counter stayed
unchanged; reference the variables mid, after (or directly
db.previousTxnID.Load()) and the channel subCh in the updated logic.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: c8ebd52e-1db7-4e52-ad27-a87a34458d56
📒 Files selected for processing (3)
client/request/select.gointernal/db/subscription_collection_filter_test.gointernal/db/subscriptions.go
📜 Review details
🧰 Additional context used
🧠 Learnings (1)
📚 Learning: 2026-03-27T08:24:16.083Z
Learnt from: islamaliev
Repo: sourcenetwork/defradb PR: 4661
File: http/middleware.go:93-93
Timestamp: 2026-03-27T08:24:16.083Z
Learning: When handling `client.ErrNotAuthorizedToPerformOperation` in the defradb repository, treat it as an unauthenticated/missing-identity condition (not insufficient permissions). Map it to HTTP 401 Unauthorized rather than 403 Forbidden; this mapping is intentional and consistent with prior behavior (pre-PR `#4661`).
Applied to files:
internal/db/subscription_collection_filter_test.goclient/request/select.gointernal/db/subscriptions.go
🔇 Additional comments (2)
client/request/select.go (1)
42-56: LGTM!Also applies to: 164-170
internal/db/subscriptions.go (1)
28-28: LGTM!Also applies to: 31-37, 51-65, 93-100
Replace the instantaneous default select with a blocking select-with-timeout so a slightly-delayed wrong-collection response can no longer slip through undetected.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## develop #4909 +/- ##
============================================
- Coverage 76.27% 54.50% -21.78%
============================================
Files 606 589 -17
Lines 46838 44884 -1954
============================================
- Hits 35724 24460 -11264
- Misses 8311 17960 +9649
+ Partials 2803 2464 -339
Flags with carried forward coverage won't be shown. Click here to find out more.
... and 319 files with indirect coverage changes Continue to review full report in Codecov by Harness.
🚀 New features to boost your workflow:
|
Relevant issue(s)
Resolves #4896
Description
This PR adds a collection-level guard at the existing docID/CID filter site so the event is dropped before any work begins.
Changes
subscriptionSelector(internal/db/subscriptions.go:24) withCheckCollectionFilter(collectionID string) bool, mirroring the existingCheckDocIDFilter/CheckCIDFiltershape.targetCollectionSettercarryingSetTargetCollectionID(id string)so the subscription handler can stamp the resolved root CollectionID at subscribe-time without polluting the subscriptionSelector contract for non-subscription consumers.*request.Select: an unexportedtargetCollectionIDfield, a setter, andCheckCollectionFilterthat returnss.targetCollectionID == "" || s.targetCollectionID == collectionID. Empty default preserves current behaviour for any caller that doesn't opt in.handleSubscription: resolveselection.Nameviadb.GetCollectionByName(ctx, name)once at subscribe-time and stamp If the collection doesn't exist the subscription request fails up-front via the existing GQL-error path.Notes
targetCollectionSetterinterface. A future selector type that implementssubscriptionSelectorwithout implementingtargetCollectionSetterfalls back to the empty-string allow-all default (i.e. current behaviour). This is intentional, we don't want unknown selector types to block legitimate subscriptions — but is worth noting .targetCollectionIDfield is written at subscribe-time and read inside the event-loop goroutine. The write happens-before thegolaunch, so no race today. A future change that moves the write past the goroutine boundary would need a lock.Tasks
How has this been tested?
New tests in
internal/db/subscription_collection_filter_test.go:TestHandleSubscription_WrongCollectionEvent_OpensNoTxn— pins that a wrong-collection event does not incrementdb.previousTxnID. The atomic counter is incremented inside everyNewTxncall (db.go:230), so a flat counter across the subscription processing window is direct evidence thatsubscriptions.go:74was never reached. Confirmed RED on develop (counter advanced 6 → 7), GREEN with the fix (counter flat).TestHandleSubscription_RightCollectionEvent_StillDelivered— control test pinning that same-collectionSpecify the platform(s) on which this was tested: